#include "server.h"
#include "hash.h"

static shm_fifo_t *g_shm_fifo = NULL;
static tpool_t *g_pool = NULL;
static hashtable_t **hash = NULL;
static int g_msgid = 0;

void server_init()
{
    // 环形队列初始化
    g_shm_fifo = shm_fifo_init(10, sizeof(packet_t));
    // 创建hash表
    hash = create_hashtable();
    // 线程池创建
    g_pool = thread_pool_create(THREAD_CNT, TASK_CNT);

    g_msgid = msg_create();
}

// 订阅
void do_subscribe(char *topic, pid_t pid)
{
    datatype_t value;
    value.pid = pid;
    strcpy(value.topic, topic);

    // 主题与进程id存储到hashmap中
    insert_data_hash(hash, topic, value);

    printf("%s,%s,%d\n", __FILE__, __FUNCTION__, __LINE__);

    // 打印hash表
    print_hash(hash);
}

// 发布
void do_publish(char *topic, pid_t pid, char *content)
{
    pid_t result[RESULT_SZ];
    int ret = 0;

    // 通过主题 获取已经订阅(存放到hash表)的进程ID
    ret = search_hash(hash, topic, result);
    if (ret == 0)
    {
        printf("do not subscribe!\n");
    }

    for (int i = 0; i < ret; i++)
    {
        // 给订阅的客户端发送内容
        msg_send(g_msgid, result[i], content);
    }
}

void task_handle(void *arg)
{
    packet_t *packet = (packet_t *)arg;
    if (packet->mode == PUBLISH)
    {
        printf("publish\n");
        // 发布
        do_publish(packet->topic, packet->pid, packet->content);
    }
    else if (packet->mode == SUBSCRIBE)
    {
        printf("subscribe\n");
        // 订阅
        do_subscribe(packet->topic, packet->pid);
    }
}

void runloop()
{
    packet_t *packet = NULL;

    for (;;)
    {
        // 分配给其他不同的线程去处理任务，如果直接定义 packet 变量，则需要进行互斥
        // 分配空间 防止产生线程资源争夺
        packet = (packet_t *)malloc(sizeof(packet_t));
        memset(packet, 0, sizeof(packet_t));

        shm_fifo_read(g_shm_fifo, packet);

        printf("%s,%s,%d--topic:%s,pid:%d,content:%s,mode:%d\n", __FILE__, __FUNCTION__, __LINE__, packet->topic, packet->pid, packet->content, packet->mode);

        // 将任务函数添加到线程池
        thread_pool_add_tasks(g_pool, task_handle, (void *)packet);
    }
}